package io.kimmking.rpcfx.config.center.zookeeper;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.kimmking.rpcfx.config.center.api.ConfigCenter;
import io.kimmking.rpcfx.config.center.api.ConfigInfo;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class ZookeeperConfigCenter implements ConfigCenter {


    private CuratorFramework client;
    private final String zookeeperServer;
    private final String namespace;

    public ZookeeperConfigCenter(String server, String namespace) {
        this.zookeeperServer = server;
        this.namespace = namespace;
        init();
    }

    public CuratorFramework getClient() {
        return client;
    }

    private void init() {
        // start zk client
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder().connectString(zookeeperServer)
                .namespace(namespace).retryPolicy(retryPolicy).build();
        client.start();

    }


    @Override
    public String readStringConfig(ConfigInfo configInfo) {
        try {
            byte[] bytes = client.getData().forPath("/"+configInfo.getServerName());
            return new String(bytes);
        } catch (Exception e) {
            System.out.println("配置读取失败:"+e.getMessage());
        }
        return null;
    }

    @Override
    public void setStringConfig(ConfigInfo configInfo, String s) {
        try {
            if ( null == client.checkExists().forPath("/" + configInfo.getServerName())) {
                client.create().withMode(CreateMode.PERSISTENT).forPath("/" + configInfo.getServerName(), new byte[0]);
            }
            client.setData().forPath("/"+configInfo.getServerName(),s.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public Map<String, String> readConfig(ConfigInfo configInfo) {
        try {
            byte[] bytes = client.getData().forPath("/"+configInfo.getServerName());
            JSONObject jsonObject = JSON.parseObject(new String(bytes));
            Map<String, String> result = new HashMap<>();
            Set<String> keySet = jsonObject.keySet();
            for (String s : keySet) {
                result.put(s, jsonObject.getString(s));
            }
            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void setConfig(ConfigInfo configInfo, Map<String, String> config) {
        try {
            if ( null == client.checkExists().forPath("/" + configInfo.getServerName())) {
                client.create().withMode(CreateMode.PERSISTENT).forPath("/" + configInfo.getServerName(), new byte[0]);
            }
            client.setData().forPath("/"+configInfo.getServerName(),JSON.toJSONString(config).getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() {
        client.close();
    }
}
